package kafka;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.Properties;

/**
 * 发送 String 的 kafka 生产者
 *
 * @author yhy
 * @date 17-5-23
 * 发送字符串的 kafka 生产者
 */
public class StringKafKaProducer implements Closeable{

    private Logger logger = LoggerFactory.getLogger(StringKafKaProducer.class);
    private Producer<String, String> producer = null;

    /**
     * kafka producer 的相应模式
     */
    public enum REQUEST_ACK {
        NO_WAIT(0),
        WAIT_LEADER(1),
        WAIT_ALL(-1);

        REQUEST_ACK(int val) {
            this.val = val;
        }

        private int val = 0;

        public int getVal() {
            return val;
        }

        public void setVal(int val) {
            this.val = val;
        }
    }

    /**
     * @param brokerList kafka 集群列表
     * @param requestAck 0, which means that the producer never waits for an acknowledgement from the broker (the same behavior as 0.7). This option provides the lowest latency but the weakest durability guarantees (some data will be lost when a server fails).
     *                   1, which means that the producer gets an acknowledgement after the leader replica has received the data. This option provides better durability as the client waits until the server acknowledges the request as successful (only messages that were written to the now-dead leader but not yet replicated will be lost).
     *                   -1, which means that the producer gets an acknowledgement after all in-sync replicas have received the data. This option provides the best durability, we guarantee that no messages will be lost as long as at least one in sync replica remains.
     */
    public StringKafKaProducer(String brokerList, REQUEST_ACK requestAck){
        Properties proPerties = new Properties();
        //此处配置的是kafka broker list
        proPerties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        //最长发送延时
        proPerties.put(ProducerConfig.LINGER_MS_CONFIG, 100);
        //发送超时事件
        proPerties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG,1000);
        //发送数据的最大size
        proPerties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,20480000);
        //最多重发次数
        proPerties.put(ProducerConfig.RETRIES_CONFIG, 3);
        //value 的序列化类
        proPerties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
        //key 的序列化类,默认 为serializer.class
        proPerties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class);
        //request.required.acks
        //0, which means that the producer never waits for an acknowledgement from the broker (the same behavior as 0.7). This option provides the lowest latency but the weakest durability guarantees (some data will be lost when a server fails).
        //1, which means that the producer gets an acknowledgement after the leader replica has received the data. This option provides better durability as the client waits until the server acknowledges the request as successful (only messages that were written to the now-dead leader but not yet replicated will be lost).
        //-1, which means that the producer gets an acknowledgement after all in-sync replicas have received the data. This option provides the best durability, we guarantee that no messages will be lost as long as at least one in sync replica remains.
        proPerties.put(ProducerConfig.ACKS_CONFIG, Integer.toString(requestAck.getVal()));

        //创建 producer
        producer = new KafkaProducer<>(proPerties);
        logger.debug("初始化 producer: brokerList={}, requestAck={}", brokerList, requestAck.toString());
    }

    /**
     * 发送数据
     *
     * @param topic   主题
     * @param message 消息
     */
    public void sendMessage(String topic, String message) {

        if (StringUtils.isEmpty(topic)) {
            logger.debug("topic为空,不发送消息.topic=:" + topic);
            return;
        }

        producer.send(new ProducerRecord<>(topic, message));
        logger.debug("发送消息到主题:{}成功 \n内容:{}", topic, message);
    }



    @Override
    public void close() throws IOException {
        producer.close();
    }


}
